package com.itbaizhan.rocketmqproducer.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;

//事务生产者
/*两阶段的提交方式，1，执行本地事务的结果，broker收不到结果，会进行回查，回查后再确认消息是重新发送还是删除*/
public class TransactionMsgProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {

        //创建事务监听器
        TransactionListener listener = new TransactionListener() {
           //执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地事务");
                //等待100秒，broker会接收不到消息，会进入下面的检查事务
                try{
                    Thread.sleep(100000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
               // return LocalTransactionState.COMMIT_MESSAGE;
                return LocalTransactionState.ROLLBACK_MESSAGE;//(测试事务回滚,消费端收不到消息)
            }
            //检查本地事务的状态
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 如果没有收到生产者发送的Half Message的响应，broker发送请求到生产者回查生产者本地事务的状态
                System.out.println("检查本地事务"+messageExt);

                return LocalTransactionState.COMMIT_MESSAGE;
               // return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        };

        //创建事务消息生产者
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("producer_grp_01");
        //设置事务端的监听器
        transactionMQProducer.setTransactionListener(listener);
        //设置nameserv地址
        transactionMQProducer.setNamesrvAddr("192.168.66.100:9876");
        transactionMQProducer.start();

        //创建消息
        Message message = null;
        message=new Message("topic_springboot_02","hello transaction message ..".getBytes(RemotingHelper.DEFAULT_CHARSET));
        //发送事务消息
        transactionMQProducer.sendMessageInTransaction(message,"{\"name\":\"zhang\"}");
    }

}
